/**
 * 
 */
package org.spacegame.server;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.Serializable;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import org.spacegame.common.ObjectReceivedEvent;
import org.spacegame.common.ObjectReceivedListener;

/**
 * The {@link ServerCommunicator} class is the main class for the server. It
 * will start a new thread in which the implementation of the server will be
 * running.
 * 
 * @author vincent
 * 
 */
public class ServerCommunicator extends Thread {

	/**
	 * The port on which the server will listen for clients.
	 */
	private final int port;

	/**
	 * The list of listeners for {@link ObjectReceivedEvent}s.
	 */
	private final List<ObjectReceivedListener> listeners = new ArrayList<ObjectReceivedListener>();

	/**
	 * The selector that will be used to listen on sockets.
	 */
	private volatile Selector selector = null;

	/**
	 * Construct a new thread in which the server will be running.
	 * 
	 * @param port
	 *            The port on which to listen for clients.
	 */
	public ServerCommunicator(final int port) {
		super("ServerCommunicatorThread");
		this.port = port;
	}

	/**
	 * Add an listener that will listen to received objects from clients.
	 * 
	 * @param listener
	 *            The listener.
	 */
	public void addObjectReceivedListener(final ObjectReceivedListener listener) {
		listeners.add(listener);
	}

	@Override
	public void run() {
		try {
			selector = Selector.open();

			synchronized (this) {
				final ServerSocketChannel serverChannel = ServerSocketChannel.open();
				serverChannel.configureBlocking(false);
				serverChannel.socket().bind(new InetSocketAddress(port));
				serverChannel.register(selector, SelectionKey.OP_ACCEPT);
				this.notifyAll();
			}

			while (true) {
				selector.select();
				final Set<SelectionKey> keys = selector.selectedKeys();
				for (final SelectionKey key : keys) {
					final SelectableChannel channel = key.channel();
					if (key.isAcceptable() && channel instanceof ServerSocketChannel) {
						final ServerSocketChannel serverSocketChannel = (ServerSocketChannel) channel;
						final SocketChannel socketChannel = serverSocketChannel.accept();
						socketChannel.configureBlocking(false);
						socketChannel.register(selector, SelectionKey.OP_READ, new LinkedBlockingQueue<ByteBuffer>());
					} else if (key.isConnectable() && channel instanceof SocketChannel) {
						final SocketChannel socketChannel = (SocketChannel) channel;
						socketChannel.finishConnect();
					} else if (key.isReadable() && channel instanceof SocketChannel) {
						final SocketChannel socketChannel = (SocketChannel) channel;
						final ByteBuffer buffer = ByteBuffer.allocate(1024);
						socketChannel.read(buffer);
						// TODO Check for closed channels.
						buffer.flip();
						final ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(buffer.array());
						final ObjectInputStream objectInputStream = new ObjectInputStream(byteArrayInputStream);
						final Serializable object = (Serializable) objectInputStream.readObject();
						final ObjectReceivedEvent event = new ObjectReceivedEvent(this, object, socketChannel.socket()
								.getRemoteSocketAddress());
						for (final ObjectReceivedListener listener : listeners) {
							listener.objectReceived(event);
						}
					} else if (key.isWritable() && channel instanceof SocketChannel) {
						final SocketChannel socketChannel = (SocketChannel) channel;
						@SuppressWarnings("unchecked")
						final BlockingQueue<ByteBuffer> buffers = (BlockingQueue<ByteBuffer>) key.attachment();
						if (buffers.isEmpty()) {
							key.interestOps(key.interestOps() & ~SelectionKey.OP_WRITE);
						} else {
							final ByteBuffer buffer = buffers.peek();
							socketChannel.write(buffer);
							if (!buffer.hasRemaining()) {
								buffers.take();
							}
						}
					}
					keys.remove(key);
				}
			}
		} catch (final Exception e) {
			e.printStackTrace();
		}
	}

	/**
	 * Send an object to all the clients.
	 * 
	 * @param object
	 *            The object to be sent.
	 * @throws IOException
	 *             There was a problem sending the object.
	 */
	public void send(final Serializable object) throws IOException {
		final ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
		final ObjectOutputStream objectOutputStream = new ObjectOutputStream(byteArrayOutputStream);
		objectOutputStream.writeObject(object);
		final byte[] bytes = byteArrayOutputStream.toByteArray();
		final ByteBuffer byteBuffer = ByteBuffer.wrap(bytes);
		final Set<SelectionKey> keys = selector.keys();
		for (final SelectionKey key : keys) {
			final SelectableChannel selectableChannel = key.channel();
			if (!(selectableChannel instanceof SocketChannel)) {
				continue;
			}
			@SuppressWarnings("unchecked")
			final BlockingQueue<ByteBuffer> buffers = (BlockingQueue<ByteBuffer>) key.attachment();
			buffers.offer(byteBuffer.asReadOnlyBuffer());
			key.interestOps(key.interestOps() | SelectionKey.OP_WRITE);
		}
		selector.wakeup();
	}

}
